package mr

import (
	"encoding/binary"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"log"
	"net"
	"net/http"
	"net/rpc"
	"os"
	"sort"
	"strconv"
	"sync"
	"time"
)

// Map functions return a slice of KeyValue.
type KeyValue struct {
	Key   string
	Value string
}

// for sorting by key.
type ByKey []KeyValue

// for sorting by key.
func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

type Slave struct {
	id      int
	state   sStatus
	mapf    func(string, string) []KeyValue
	reducef func(string, []string) string
}

func LocalFileWrite(filepath string, content []byte) error {
	file, err := os.OpenFile(filepath+"-tmp", os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0644)
	if err != nil {
		fmt.Println("无法打开文件:", err)
		return err
	}

	_, err = file.Write(content)
	if err != nil {
		fmt.Println("写入文件失败:", err)
		return err
	}
	file.Close()

	err = os.Rename(filepath+"-tmp", filepath)

	return err
}

func LocalFilesRead(filepaths []string) ([][]byte, error) {
	var wg sync.WaitGroup
	var mutex sync.Mutex
	wg.Add(len(filepaths))
	contents := make([][]byte, len(filepaths))
	for _, filename := range filepaths {
		go func(filename string) error {
			defer wg.Done()
			file, err := os.Open(filename)
			if err != nil {
				log.Fatalf("cannot open %v\n", filename)
			}
			content, err := ioutil.ReadAll(file)
			if err != nil {
				log.Fatalf("cannot read %v\n", filename)
			}
			file.Close()
			mutex.Lock()
			contents = append(contents, content)
			mutex.Unlock()
			return nil
		}(filename)
	}
	wg.Wait()
	return contents, nil
}

func (this *Slave) Map(args *MapArgs, reply *MapReply) error {

	contents, err := LocalFilesRead(args.Filenames)
	if err != nil {
		log.Fatalf("cannot open %v\n", args.Filenames)
	}

	kva := make([]KeyValue, 0)

	for i, filename := range args.Filenames {
		kva = append(kva, this.mapf(filename, string(contents[i]))...)
	}

	sort.Sort(ByKey(kva))

	i := 0
	for i < len(kva) {
		j := i + 1
		for j < len(kva) && kva[j].Key == kva[i].Key {
			j++
		}
		values := []string{}
		for k := i; k < j; k++ {
			values = append(values, kva[k].Value)
		}

		filepath := "mr-" + strconv.Itoa(this.id) + "-" + kva[i].Key
		reply.Buckets[kva[i].Key] = filepath

		content, err := json.Marshal(values)
		if err != nil {
			return err
		}
		go LocalFileWrite(filepath, content)
		i = j
	}

	return nil
}

func (this *Slave) Reduce(args *ReduceArgs, reply *ReduceReply) error {
	outputs := make([]string, len(args.KeysFiles))
	for k, v := range args.KeysFiles {
		contents, err := LocalFilesRead(v)
		if err != nil {
			log.Fatalf("cannot open %v\n", v)
		}
		value := make([]string, 0)
		for _, c := range contents {
			t := make([]string, 0)
			json.Unmarshal(c, t)
			value = append(value, t...)
		}

		output := this.reducef(k, value)
		outputs = append(outputs, output)
	}
	reply.N = outputs

	return nil
}

func (this *Slave) server() {
	go this.heartBeat()

	sockname := getSockName(os.Getpid())
	os.Remove(sockname)
	l, e := net.Listen("unix", sockname)
	if e != nil {
		log.Fatalln("listen error:", e)
	}

	rpc.Register(this)
	rpc.HandleHTTP()
	go http.Serve(l, nil)
}

func (this *Slave) heartBeat() {
	sockname := getSockName(-1)
	c, err := net.Dial("unix", sockname)
	if err != nil {
		log.Fatalln("dialing:", err)
	}
	defer c.Close()

	for {
		buf := make([]byte, 4)
		binary.BigEndian.PutUint32(buf, uint32(this.id))
		_, errw := c.Write(buf)
		if errw != nil {
			log.Print(err)
			return
		}
		_, errr := c.Read(buf)
		if errr != nil {
			panic(err)
		}
		fmt.Println(buf)
		time.Sleep(time.Second * 12)
	}
}

// main/mrworker.go calls this function.
func Worker(mapf func(string, string) []KeyValue, reducef func(string, []string) string) {
	s := Slave{id: os.Getpid(), state: Available, mapf: mapf, reducef: reducef}
	s.server()
	for {

	}
}
